package com.thiswhyme.republish.rabbitmq.core;

import cn.hutool.core.date.DateUtil;
import com.thiswhyme.republish.common.RedisKey;
import com.thiswhyme.republish.rabbitmq.core.domain.MessageInfo;
import com.thiswhyme.republish.rabbitmq.core.domain.PublishStatus;
import com.thiswhyme.republish.rabbitmq.datasource.HistoryMessageInfoRepo;
import com.thiswhyme.republish.rabbitmq.datasource.MessageInfoRepo;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.CollectionUtils;

import java.sql.Timestamp;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * 历史消息对象维护服务
 * Created by thiswhyme on 2021-03-29.
 */
public class HistoryMessageInfoService {

    public static final Logger log = LoggerFactory.getLogger(HistoryMessageInfoService.class);

    public static final int BATCH_SIZE = 100;

    @Value("${spring.application.name}")
    private String applicationName;

    private MessageInfoRepo messageInfoRepo;
    private HistoryMessageInfoRepo historyMessageInfoRepo;
    private RedissonClient redissonClient;

    public HistoryMessageInfoService(MessageInfoRepo messageInfoRepo, HistoryMessageInfoRepo historyMessageInfoRepo, RedissonClient redissonClient) {
        this.messageInfoRepo = messageInfoRepo;
        this.historyMessageInfoRepo = historyMessageInfoRepo;
        this.redissonClient = redissonClient;
    }

    /**
     * 每天凌晨1点，迁移前一天之前的已完成消息到历史表
     */
    @Scheduled(cron="0 0 1 * * ?")
    public void autoTransfer() {
        String lockKey = RedisKey.genKey(applicationName, RedisKey.MESSAGE_TRANSFER_LOCK);
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(new Date());
        calendar.add(Calendar.DATE, -1);
        Date date = DateUtil.endOfDay(calendar.getTime());
        RLock lock = redissonClient.getLock(lockKey);
        try {
            if(lock.tryLock(300, TimeUnit.SECONDS)) {
                log.info("【历史消息迁移定时器】获取消息迁移锁成功，开始执行迁移{}之前已完成的消息", date);
                Integer count = transferFinishedMsgsToHistory(date.toInstant());
                log.info("【历史消息迁移定时器】结束执行迁移{}之前已完成的消息，共{}条", date, count);
            } else {
                log.info("【历史消息迁移定时器】消息迁移锁已被其他实例占用，本实例跳过");
            }
        } catch (Exception e) {
            log.error("【历史消息迁移定时器】消息迁移操作失败", e);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 迁移指定时间之前的数据到历史表
     * @param instant
     * @return
     */
    private int transferFinishedMsgsToHistory(Instant instant) {
        Timestamp limitTime = Timestamp.from(instant);
        int total = 0;
        List<MessageInfo> messageInfos = messageInfoRepo.selectByCreateTime(Arrays.asList(PublishStatus.SEND_SUCCESS.getValue(),
                PublishStatus.FAIL.getValue(), PublishStatus.NONE.getValue()), limitTime, 0, BATCH_SIZE);
        Map<String, List<MessageInfo>> messageMap = groupByCreateMonth(messageInfos);
        while (!CollectionUtils.isEmpty(messageMap)) {
            total += messageInfos.size();
            messageMap.forEach((suffix, messages) -> {
                historyMessageInfoRepo.batchTransfer(messages, suffix);
            });
            messageInfos = messageInfoRepo.selectByCreateTime(Arrays.asList(PublishStatus.SEND_SUCCESS.getValue(), PublishStatus.FAIL.getValue(),
                    PublishStatus.NONE.getValue()), limitTime, 0, BATCH_SIZE);
            messageMap = groupByCreateMonth(messageInfos);
        }
        return total;
    }

    /**
     * 消息按月分组
     * @param messageInfos
     * @return
     */
    private Map<String, List<MessageInfo>> groupByCreateMonth(List<MessageInfo> messageInfos) {
        if(CollectionUtils.isEmpty(messageInfos)) {
            return Collections.EMPTY_MAP;
        }
        Map<String, List<MessageInfo>> results = new HashMap<>();
        messageInfos.forEach(messageInfo -> {
            String key = historyMessageInfoRepo.generateTableNameSuffix(messageInfo.getCreateTime());
            List<MessageInfo> msgs = Optional.ofNullable(results.get(key)).orElseGet(() -> new ArrayList<>());
            msgs.add(messageInfo);
            results.put(key, msgs);
        });
        return results;
    }
}
